package com.landcode.land.data.canal.manager;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.landcode.land.data.canal.config.CanalServerConfig;
import com.landcode.land.data.canal.util.CanalUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.SystemUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Date;

@Slf4j
public abstract class AbstractCanalManager {

    @Autowired
    protected CanalServerConfig config;

    protected CanalConnector canalConnector;

    protected final static String dateFormat = "yyy-MM-dd HH:mm:ss";

    protected final static String SEP = SystemUtils.LINE_SEPARATOR;

    public abstract void process();

    public abstract void synData(Message message) throws Exception;

    public final CanalConnector getCanalConnector() {
        log.info("------canal server:" + config.getServers());
        log.info("------canal destination:" + config.getDestination());
        if (config.isCluster()) {
            canalConnector = CanalConnectors.newClusterConnector(config.getServers(), config.getDestination(), config.getUsername(), config.getPassword());
        } else {
            if (canalConnector == null) {
                if (config.getServers() != null) {
                    InetSocketAddress inetSocketAddress = CanalUtil.stringToAddress(config.getServers());
                    canalConnector = CanalConnectors.newSingleConnector(inetSocketAddress, config.getDestination(), config.getUsername(), config.getPassword());
                }
            }
        }
        return canalConnector;
    }

    protected void printMsg(Message message, long batchId, int size) {
/*        long memsize = 0;
        for (CanalEntry.Entry entry : message.getEntries()) {
            memsize += entry.getHeader().getEventLength();
        }*/

        long memsize = message.getEntries().stream().mapToLong(i -> i.getHeader().getEventLength()).sum();

        String startPosition = null;
        String endPosition = null;
        if (!CollectionUtils.isEmpty(message.getEntries())) {
            startPosition = buildPositionForDump(message.getEntries().get(0));
            endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
        }

        String contextFormat = SEP + "****************************************************" + SEP;
        contextFormat += "* Batch Id: [{}] ,count : [{}] , memsize : [{}] , Time : {}" + SEP;
        contextFormat += "* Start : [{}] " + SEP;
        contextFormat += "* End : [{}] " + SEP;
        contextFormat += "****************************************************" + SEP;

        SimpleDateFormat format = new SimpleDateFormat(dateFormat);
        log.info(contextFormat, new Object[]{batchId, size, memsize, format.format(new Date()), startPosition,
                endPosition});
    }

    protected String buildPositionForDump(CanalEntry.Entry entry) {
        long time = entry.getHeader().getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(dateFormat);
        return entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
                + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
    }
}
